package common

import (
	"context"
	"errors"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/sirupsen/logrus"
)

type MQTT struct {
	option     *mqtt.ClientOptions
	prefix     string        //topic名称前缀
	log        *logrus.Entry //日志
	conn       mqtt.Client
	clientSubs []*mqttSub //客户端
}

type mqttSub struct {
	ctx        context.Context
	cancelFunc context.CancelFunc
	topic      string
	handler    mqtt.MessageHandler
}

// NewMQTT MQTT协议通讯
func NewMQTT(addr string, username string, password string, prefix string) (*MQTT, error) {
	optione := mqtt.NewClientOptions()
	optione.CleanSession = true
	optione.AddBroker(addr)
	optione.SetUsername(username)
	optione.SetPassword(password)
	ret := &MQTT{
		option: optione,
		prefix: prefix,
	}
	err := ret.Init(false)
	return ret, err
}

func (a *MQTT) Init(reload bool) error {
	if !reload && a.conn != nil {
		return nil
	}
	a.Close()
	a.conn = mqtt.NewClient(a.option)
	token := a.conn.Connect()
	token.Wait()
	err := token.Error()
	if err != nil {
		return err
	}
	for _, v := range a.clientSubs { //重新监听
		a.addSubscribe(v)
	}
	return nil
}

func (a *MQTT) Reload() {
	a.Init(true)
}

// NewSubscribe 创建客户端
func (a *MQTT) NewSubscribe(topic string, handler mqtt.MessageHandler) error {
	for _, v := range a.clientSubs {
		if v.topic == topic {
			return errors.New("该话题已有订阅")
		}
	}
	req := &mqttSub{
		topic:   topic,
		handler: handler,
	}
	err := a.addSubscribe(req)
	if err != nil {
		return err
	}
	a.clientSubs = append(a.clientSubs, req)
	return nil
}

func (a *MQTT) addSubscribe(req *mqttSub) error {
	a.Init(false)
	if a.conn == nil {
		return errors.New("nats未连接")
	} else if req.handler == nil {
		return errors.New("处理函数空")
	}
	tk := a.conn.Subscribe(a.prefix+req.topic, 0x00, req.handler)
	tk.Wait()
	return tk.Error()
}

// Put 推送一个消息
func (a *MQTT) Put(topic string, msg []byte) error {
	a.Init(false)
	tk := a.conn.Publish(a.prefix+topic, 0x00, false, msg)
	tk.Wait()
	return tk.Error()
}

func (a *MQTT) Close() {
	if a.conn != nil {
		topics := make([]string, len(a.clientSubs))
		for i, v := range a.clientSubs {
			topics[i] = v.topic
		}
		a.conn.Unsubscribe(topics...)
		a.conn.Disconnect(30)
		a.conn = nil
	}
}
